package org.lzfto.flink.demo.business.flinkTest.flinkObjects.sink;

import java.io.IOException;

import org.apache.flink.api.connector.sink2.SinkWriter;
import org.lzfto.flink.demo.common.result.PubResult;
import org.lzfto.flink.demo.config.websocket.MyWebSocketHandler;

import com.alibaba.fastjson2.JSON;

/**
*自定义的websocket输出类
*<br/>如果是生产环境，应该更加严谨一些。此处仅作演示
*@author lzfto
*@创建时间 2025年6月18日 11:37:40
*/
public class CustWebSocketSinkWriter<T> implements SinkWriter<T> {

	private String  websocketClientName;
	private String  messagePrefix;

	
	public CustWebSocketSinkWriter(String websocketClientName,String messagePrefix) {
		super();
		this.websocketClientName=websocketClientName;
		this.messagePrefix=messagePrefix;
	}
	
	@Override
	public void close() throws Exception {
		//无需关闭
	}

	@Override
	public void write(T element, Context context) throws IOException, InterruptedException {
		// element应该有合适的toString()方法
		
		if (element!=null) {
			if (messagePrefix.equals("")) {
				PubResult<Object> message=PubResult.ok_data(element).withReceiver("flink").withDataIsObject(true);
				MyWebSocketHandler.sendTxtTo(JSON.toJSONString(message),websocketClientName);	
			}
			else {
				PubResult<Object> message=PubResult.ok_data(messagePrefix+element.toString()).withReceiver("flink").withDataIsObject(true);
				MyWebSocketHandler.sendTxtTo(JSON.toJSONString(message),websocketClientName);
			}
		}
		else {
			PubResult<Object> message=PubResult.error("没有收到合适的数据").withReceiver("flink");
			MyWebSocketHandler.sendTxtTo(JSON.toJSONString(message),websocketClientName);
		}
	}

	@Override
	public void flush(boolean endOfInput) throws IOException, InterruptedException {
       //无需flush		
	}

}
